package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.BrowsingBean;
import com.atguigu.edu.realtime.beans.BrowsingBehavior;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

public class DwsBrowsingBehavior {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        env.enableCheckpointing(5000L);

        String topic = "dwd_traffic_page_log";
        String groupId = "DwsBrowsingBehavior";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        SingleOutputStreamOperator<JSONObject> jsonObjDs = kafkaDs.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> filteredDs = jsonObjDs.process(
                new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        String aa = value.getJSONObject("page").getString("page_id");
                        if (aa.equals("home") || aa.equals("course_detail")) {
                            out.collect(value);
                        }
                    }
                }
        );
        SingleOutputStreamOperator<JSONObject> withWatermarkDs = filteredDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }
                        )
        );
        SingleOutputStreamOperator<BrowsingBehavior> mapDs = withWatermarkDs.map(
                new MapFunction<JSONObject, BrowsingBehavior>() {
                    @Override
                    public BrowsingBehavior map(JSONObject value) throws Exception {
                        String uid = value.getJSONObject("common").getString("uid");
                        String page_id = value.getJSONObject("page").getString("page_id");
                        Long ts = value.getLong("ts");
                        return new BrowsingBehavior(uid, page_id, ts);
                    }
                }
        );
        //mapDs.print("^^^");
        KeyedStream<BrowsingBehavior, String> keyedDs = mapDs.keyBy(aa -> aa.getPage_id());
        KeyedStream<BrowsingBehavior, String> keyedAaDs = keyedDs.keyBy(aa -> aa.getUid());
        SingleOutputStreamOperator<BrowsingBehavior> firstDataDs = keyedAaDs.process(
                new KeyedProcessFunction<String, BrowsingBehavior, BrowsingBehavior>() {
                    private ValueState<String> lastDatestate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stringValueStateDescriptor
                                = new ValueStateDescriptor<>("lastDatestate", String.class);
                        stringValueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastDatestate = getRuntimeContext().getState(stringValueStateDescriptor);
                    }

                    @Override
                    public void processElement(BrowsingBehavior value, KeyedProcessFunction<String, BrowsingBehavior, BrowsingBehavior>.Context ctx, Collector<BrowsingBehavior> out) throws Exception {
//按状态看是不是今天此用户的第一条数据，是的话才往下传
//获取上次加购日期
                        String lastDate = lastDatestate.value();
                        Long ts = value.getTs();
                        String date = DateFormatUtil.toDate(ts);
                        LocalDate now = LocalDate.now();
                        String format = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
                        if ((StringUtils.isEmpty(lastDate) && date.equals(format)) || (StringUtils.isNotEmpty(lastDate) && !lastDate.equals(format))) {
                            out.collect(value);
                            lastDatestate.update(date);
                        }
                    }
                }
        );
       //firstDataDs.print("----");



        KeyedStream<BrowsingBehavior, String> keyedPage = firstDataDs.keyBy(aa -> aa.getPage_id());
        AllWindowedStream<BrowsingBehavior, TimeWindow> withWindowDs = keyedPage.windowAll(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(3)));


    /*    withWindowDs.reduce(
                new ReduceFunction<BrowsingBehavior>() {
                    @Override
                    public BrowsingBehavior reduce(BrowsingBehavior value1, BrowsingBehavior value2) throws Exception {
                        return null;
                    }
                }
        )*/







        SingleOutputStreamOperator<BrowsingBean> aggregateDs = withWindowDs.process(
                new ProcessAllWindowFunction<BrowsingBehavior, BrowsingBean, TimeWindow>() {

                    @Override
                    public void process(ProcessAllWindowFunction<BrowsingBehavior, BrowsingBean, TimeWindow>.Context context, Iterable<BrowsingBehavior> elements, Collector<BrowsingBean> out) throws Exception {
                        Long sum = 0L;
                        String page_id = null;
                        for (BrowsingBehavior element : elements) {
                            sum++;
                            page_id = element.getPage_id();
                        }
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());
                        out.collect(new BrowsingBean(stt, edt, curDate, page_id, sum));
                    }
                }
        );

        aggregateDs.addSink(new DorisSinkFunction<BrowsingBean>("dws_user_browsing_behavior"));
        aggregateDs.print(">>>");












       /* OutputTag<BrowsingBehavior> firstPage = new OutputTag<BrowsingBehavior>("firstPage", TypeInformation.of(BrowsingBehavior.class));
        OutputTag<BrowsingBehavior> coursePage = new OutputTag<BrowsingBehavior>("coursePage", TypeInformation.of(BrowsingBehavior.class));
        SingleOutputStreamOperator<BrowsingBehavior> lissPageDS = withWatermarkDs.process(
                new ProcessFunction<JSONObject, BrowsingBehavior>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, BrowsingBehavior>.Context ctx, Collector<BrowsingBehavior> out) throws Exception {
                        String uid = value.getJSONObject("common").getString("uid");
                        String page_id = value.getJSONObject("page").getString("page_id");
                        Long ts = value.getLong("ts");
                        if ("home".equals(page_id)) {
                            ctx.output(firstPage, new BrowsingBehavior(uid, page_id, ts));
                        }
                        if ("course_detail".equals(page_id)) {
                            ctx.output(coursePage, new BrowsingBehavior(uid, page_id, ts));
                        }

                    }
                }
        );
        SideOutputDataStream<BrowsingBehavior> firstPageDs = lissPageDS.getSideOutput(firstPage);
        SideOutputDataStream<BrowsingBehavior> coursePageDS = lissPageDS.getSideOutput(coursePage);*/



env.execute();
    }
}
